"""
Case Type   : 函数支持指定并行相关参数
Case Name   : 单个游标入参场景--函数声明为parallel_enable,但不为SRF函数或者管道函数，
              入参游标是由游标表达式声明的游标，顶层是stream算子，且支持并行，
              函数调用放在from子句之后
Create At   : 2024/10/12
Owner       : @kyleze
Description :
    1.创建测试表
    2.函数声明为parallel_enable,但不为SRF函数或者管道函数
    3.调用函数是否支持并行
    4.清理环境
Expect      :
    1.成功
    2.成功
    3.不支持
    4.成功
History     :
"""
import os
import unittest

from testcase.utils.CommonSH import CommonSH
from testcase.utils.Logger import Logger
from testcase.utils.Constant import Constant


class ParallelEnable(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.table_name = 't_parallel_test007'
        self.cursor_name = 'c_parallel_test007'
        self.type_name = 'type_parallel_test007'
        self.function_name = 'f_parallel_test007'

    def test_paralle(self):
        step = '----step1:创建测试表 expect:成功----'
        self.log.info(step)
        create_sql = f"drop table if exists {self.table_name} cascade;" \
                     f"create table {self.table_name} (employee_id " \
                     f"number(10),department_id number, first_name " \
                     f"varchar2(30), last_name varchar2(30), email " \
                     f"varchar2(30), phone_number varchar2(30));" \
                     f"insert into {self.table_name} values " \
                     f"(generate_series(1, 9000000), 60, 'abc', 'def', " \
                     f"'123', '123');analyze {self.table_name};" \
                     f"create type {self.type_name} as (" \
                     f"    employee_id numeric(10,0)," \
                     f"    department_id numeric," \
                     f"    first_name character varying(30)," \
                     f"    last_name character varying(30)," \
                     f"    email character varying(30)," \
                     f"    phone_number character varying(30));"
        create_res = self.primary_sh.execut_db_sql(create_sql)
        self.log.info(create_res)
        self.assertIn(
            self.constant.CREATE_TABLE_SUCCESS, create_res, '执行失败' + step)
        self.assertIn(
            self.constant.INSERT_SUCCESS_MSG, create_res, '执行失败' + step)
        self.assertIn(
            self.constant.CREATE_TYPE_SUCCESS_MSG, create_res, '执行失败' + step)

        step = '----step2:函数声明为parallel_enable,但不为SRF函数或者管道函数' \
               'expect:成功----'
        self.log.info(step)
        fun_sql = f'drop function if exists {self.function_name};' \
                  f'create or replace function {self.function_name} ' \
                  f'(p sys_refcursor) return int parallel_enable ' \
                  f'(partition p by hash(employee_id)) is' \
                  f'    out_rec {self.type_name} := {self.type_name}(null, ' \
                  f'null, null, null, null, null);' \
                  f'    res int := 0;' \
                  f'begin' \
                  f'    loop' \
                  f'        fetch from p into out_rec.employee_id, ' \
                  f'out_rec.department_id, out_rec.first_name, ' \
                  f'out_rec.last_name, out_rec.email, out_rec.phone_number;' \
                  f'        exit when p%notfound;' \
                  f'        res := res + 1;' \
                  f'    end loop;' \
                  f'    return res;' \
                  f'end {self.function_name};'
        fun_res = self.primary_sh.execut_db_sql(fun_sql)
        self.log.info(fun_res)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG,
                      fun_res, '执行失败' + step)

        step = '----step3:调用函数是否支持并行 expect:不支持----'
        self.log.info(step)
        tra_sql = f'set query_dop = 10;' \
                  f'explain (costs off) select * from {self.table_name};' \
                  f'explain (costs off) select * from  ' \
                  f'{self.function_name}(cursor (select * from ' \
                  f'{self.table_name}));'
        transaction_res = self.primary_sh.execut_db_sql(tra_sql)
        self.log.info(transaction_res)
        self.assertEqual(
            1, transaction_res.count("dop: 1/10"), '执行失败' + step)

    def tearDown(self):
        self.log.info('----清理环境----')
        table_res = \
            self.primary_sh.execut_db_sql(f'drop table {self.table_name};')
        self.log.info(table_res)
        fun_res = self.primary_sh.execut_db_sql(
            f'drop function {self.function_name};')
        self.log.info(fun_res)
        type_res = self.primary_sh.execut_db_sql(f'drop type {self.type_name};')
        self.log.info(type_res)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, table_res, '删表失败')
        self.assertIn(self.constant.DROP_FUNCTION_SUCCESS_MSG,
                      fun_res, '删表失败')
        self.assertIn(self.constant.DROP_TYPE_SUCCESS_MSG, type_res, '删表失败')
        self.log.info(f'----{os.path.basename(__file__)} end----')
